package tool

import akka.actor.ActorSystem
import akka.actor.typed.{ActorRef, Behavior, Terminated}
import akka.actor.typed.scaladsl.Behaviors
import akka.actor.typed.scaladsl.adapter.ClassicActorSystemOps
import akka.stream.OverflowStrategy
import akka.stream.scaladsl.{Flow, Keep, Sink, Source}
import akka.stream.typed.scaladsl.{ActorSink, ActorSource}

/**
 * Created by yz on 7/7/2021
 */
object TypedActorFlow {

  sealed trait EndOfStream
  case object Success extends EndOfStream
  case object Failure extends EndOfStream

  def actorRef[In, Out](
                         behavior: ActorRef[Out] => Behavior[In],
                         bufferSize: Int = 16,
                         overflowStrategy: OverflowStrategy = OverflowStrategy.dropNew
                       )(implicit actorSystem: ActorSystem): Flow[In, Out, _] = {
    val (outActor, publisher) = ActorSource
      .actorRef[Out](
        {
          case _ if false =>
        }: PartialFunction[Out, Unit], {
          case e: Any if false => new Exception(e.toString)
        }: PartialFunction[Any, Throwable],
        bufferSize,
        overflowStrategy
      )
      .toMat(Sink.asPublisher(false))(Keep.both)
      .run()

    //ActorRef[Out] out js端 Source
    //ActorRef[In] in play端 Sink

    val sink = Flow[In]
      .map(Right[EndOfStream, In])
      .to(
        ActorSink.actorRef[Either[EndOfStream, In]](
          actorSystem.spawnAnonymous(
            Behaviors.setup[Either[EndOfStream, In]] { context =>
              val flowActor = context.spawn(behavior(outActor), "flowActor")
              context.watch(flowActor)

              Behaviors
                .receiveMessage[Either[EndOfStream, In]] {
                  case Right(in) =>
                    flowActor ! in
                    Behaviors.same
                  case Left(_) =>
                    context.stop(flowActor)
                    Behaviors.same
                }
                .receiveSignal {
                  case (_, Terminated(_)) =>
                    Behaviors.stopped
                }
            }
          ),
          Left(Success),
          _ => Left(Failure)
        )
      )

    Flow.fromSinkAndSource(
      sink,
      Source.fromPublisher(publisher)
    )
  }

}
